package com.bw.app.dwd;

import com.bw.utils.MyKafkaUtil;
import lombok.val;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author y3068
 * @Classname Test1
 * @Description TODO
 * @Date 2025/7/1 17:05
 * @Version V1.0
 */

public class Test1 {

    public static void main(String[] args) throws Exception {
//你数据能写到卡夫卡？ 可以  叫啥  就执行这个test1出不来啊  对 就这个  log和五连查都有数据，就test没有
//        你这test是干什么的，准备把日志和业务用shopID连起来    业务练的哪个表
//        那就是业务和日志的shopID  都是随机生产的呗 是的
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        val topic = "dwd_traffic_page_log";
        val groupId = "group_id";

        tEnv.executeSql(
                "create table dwd_order_pre( " +
                        "    `id` string, " +
                        "    `order_id` string, " +
                        "    `shop_id` string, " +
                        "    `sku_id` string, " +
                        "    `sku_name` string, " +
                        "    `order_price` string, " +
                        "    `sku_num` string, " +
                        "    `create_time` string, " +
                        "    `source_type_id` string, " +
                        "    `source_type_name` string, " +
                        "    `source_id` string, " +
                        "    `split_total_amount` string, " +
                        "    `split_activity_amount` string, " +
                        "    `split_coupon_amount` string, " +
                        "    `consignee` string, " +
                        "    `consignee_tel` string, " +
                        "    `total_amount` string, " +
                        "    `order_status` string, " +
                        "    `user_id` string, " +
                        "    `payment_way` string, " +
                        "    `delivery_address` string, " +
                        "    `order_comment` string, " +
                        "    `out_trade_no` string, " +
                        "    `trade_body` string, " +
                        "    `operate_time` string, " +
                        "    `expire_time` string, " +
                        "    `process_status` string, " +
                        "    `tracking_no` string, " +
                        "    `parent_order_id` string, " +
                        "    `province_id` string, " +
                        "    `activity_reduce_amount` string, " +
                        "    `coupon_reduce_amount` string, " +
                        "    `original_total_amount` string, " +
                        "    `feight_fee` string, " +
                        "    `feight_fee_reduce` string, " +
                        "    `refundable_time` string, " +
                        "    `order_detail_activity_id` string, " +
                        "    `activity_id` string, " +
                        "    `activity_rule_id` string, " +
                        "    `order_detail_coupon_id` string, " +
                        "    `coupon_id` string, " +
                        "    `coupon_use_id` string, " +
                        "    `type` string, " +
                        "    `old` map<string,string>, " +
                        "    row_op_ts TIMESTAMP_LTZ(3) "+
                        ")" + MyKafkaUtil.getKafkaDDL("dwd_trade_order_pre_process","aksdghjasd"));
//        tEnv.executeSql("select * from dwd_order_pre").print();


        tEnv.executeSql("" +
                "create table page_log(\n" +
                "   `common` map<string,string>,\n" +
                "   `page` map<string,string>\n" +
                ")" + MyKafkaUtil.getKafkaDDL(topic,groupId));

        Table pageLog = tEnv.sqlQuery("" +
                "select\n" +
                "   common['uid'] uid,\n" +
                "   common['mid_type'] mid_type,\n" +
                "   common['md'] md,\n" +
                "   common['mid'] mid,\n" +
                "   page['page_id'] page_id,\n" +
                "   page['shop_id'] shop_id\n" +
                "from page_log");
        tEnv.createTemporaryView("page_log",pageLog);
//        tEnv.executeSql("select * from page_log").print();


        tEnv.executeSql(
                "create table dwd_wide( " +
                        "    `uid` string, " +
                        "    `mid_type` string, " +
                        "    `md` string, " +
                        "    `mid` string, " +
                        "    `page_id` string, " +
                        "    `shop_id` string, " +
                        "    `id` string, " +
                        "    `order_id` string, " +
                        "    `sku_id` string, " +
                        "    `sku_name` string, " +
                        "    `order_price` string, " +
                        "    `sku_num` string, " +
                        "    `create_time` string, " +
                        "    `source_type_id` string, " +
                        "    `source_type_name` string, " +
                        "    `source_id` string, " +
                        "    `split_total_amount` string, " +
                        "    `split_activity_amount` string, " +
                        "    `split_coupon_amount` string, " +
                        "    `consignee` string, " +
                        "    `consignee_tel` string, " +
                        "    `total_amount` string, " +
                        "    `order_status` string, " +
                        "    `user_id` string, " +
                        "    `payment_way` string, " +
                        "    `delivery_address` string, " +
                        "    `order_comment` string, " +
                        "    `out_trade_no` string, " +
                        "    `trade_body` string, " +
                        "    `operate_time` string, " +
                        "    `expire_time` string, " +
                        "    `process_status` string, " +
                        "    `tracking_no` string, " +
                        "    `parent_order_id` string, " +
                        "    `province_id` string, " +
                        "    `activity_reduce_amount` string, " +
                        "    `coupon_reduce_amount` string, " +
                        "    `original_total_amount` string, " +
                        "    `feight_fee` string, " +
                        "    `feight_fee_reduce` string, " +
                        "    `refundable_time` string, " +
                        "    `order_detail_activity_id` string, " +
                        "    `activity_id` string, " +
                        "    `activity_rule_id` string, " +
                        "    `order_detail_coupon_id` string, " +
                        "    `coupon_id` string, " +
                        "    `coupon_use_id` string, " +
                        "    `type` string, " +
                        "    row_op_ts TIMESTAMP_LTZ(3) "+
                        ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_wide"));


        Table result_table = tEnv.sqlQuery("select\n" +
                "l.uid as uid,\n" +
                "l.mid_type as mid_type,\n" +
                "l.md as md,\n" +
                "l.mid as mid,\n" +
                "l.page_id as page_id,\n" +
                "l.shop_id as shop_id,\n" +
                "op.id ,\n" +
                "op.order_id,\n" +
                "op.sku_id,\n" +
                "op.sku_name,\n" +
                "op.order_price,\n" +
                "op.sku_num,\n" +
                "op.create_time,\n" +
                "op.source_type_id,\n" +
                "op.source_type_name,\n" +
                "op.source_id,\n" +
                "op.split_total_amount,\n" +
                "op.split_activity_amount,\n" +
                "op.split_coupon_amount,\n" +
                "op.consignee,\n" +
                "op.consignee_tel,\n" +
                "op.total_amount,\n" +
                "op.order_status,\n" +
                "op.user_id,\n" +
                "op.payment_way,\n" +
                "op.delivery_address,\n" +
                "op.order_comment,\n" +
                "op.out_trade_no,\n" +
                "op.trade_body,\n" +
                "op.operate_time,\n" +
                "op.expire_time,\n" +
                "op.process_status,\n" +
                "op.tracking_no ,\n" +
                "op.parent_order_id,\n" +
                "op.province_id,\n" +
                "op.activity_reduce_amount ,\n" +
                "op.coupon_reduce_amount ,\n" +
                "op.original_total_amount ,\n" +
                "op.feight_fee ,\n" +
                "op.feight_fee_reduce ,\n" +
                "op.refundable_time ,\n" +
                "op.order_detail_activity_id ,\n" +
                "op.activity_id ,\n" +
                "op.activity_rule_id ,\n" +
                "op.order_detail_coupon_id ,\n" +
                "op.coupon_id ,\n" +
                "op.coupon_use_id ,\n" +
                "op.type ,\n" +
                "current_row_timestamp() row_op_ts\n" +
                "from dwd_order_pre op\n" +
                "join page_log l on op.shop_id = l.shop_id");
        tEnv.createTemporaryView("result_table1",result_table);
//        tEnv.executeSql("select * from result_table1").print();

        tEnv.executeSql("insert into dwd_wide select * from result_table1");


//        env.execute();
    }
}
